package com.baishancloud.log.fm

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.env.StreamEnv
import com.baishancloud.log.common.sink.SinkUtil
import com.baishancloud.log.common.source.SourceUtil
import com.baishancloud.log.common.util.{LogFormatUtil, TimeUtil}
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 *
 * @author ziqiang.wang 
 * @date 2021/12/1 16:20
 */
object SingleMachineStatisticsEs extends Serializable {

  def main(args: Array[String]): Unit = {
    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    println("输入参数为：" + parameterTool.toMap.toString)

    StreamEnv.builder(parameterTool).build()
    val luban1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-金华-luban").uid("7c8d6620-2918-4a5c-bc1f-77e84c8cbabd")
    val luban2: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-喜鹊-luban").uid("18536c2d-718c-4b8a-994b-d396bc877c8e")
    val luban3: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-石家庄-luban").uid("588f79c5-9117-4345-8af6-01f92ae94328")
    val luban: DataStream[(String, String)] =
      luban1.union(luban2, luban3)
        .rebalance
        .map(("LUBAN", _)).name("luban").uid("baebd4b9-ac88-4802-86b3-4dd1ae3eef09")

    val blb1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-金华-blb").uid("17b20156-b8b6-4a47-a5dd-1f7edeacacdd")
    val blb2: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-喜鹊-blb").uid("f8c7c6a5-0e1b-475c-a717-e684380e2fb7")
    val blb: DataStream[(String, String)] =
      blb1.union(blb2)
        .rebalance
        .map(("BLB", _)).name("blb").uid("d754b957-2cfb-4cc9-9dc4-24fe52ae129c")

    luban.union(blb)
      .map(x => (x._1, LogFormatUtil.originParse(x._2))).name("Origin").uid("652d99e8-9ce2-42fe-ace-ec5354dd8f2a")
      .filter(_._2 != null).name("!=null").uid("644494f1-ded9-4433-9491-5783a807f1a")
      .filter(new LateDataFilter(parameterTool)).name("lateDataFilter").uid("e3826914-dbb5-43b5-aaa0-278eda6fe9ce")
      .map(new LogParse).name("logParse").uid("8c749a-4452-4006-a6d3-48ef81d5629")
      .keyBy(_.getKey)
      .window(TumblingProcessingTimeWindows.of(Time.minutes(parameterTool.getLong(windowTime, 2))))
      .reduce((x, y) => Log(x.getKey, LogValue(x.request + y.request, x.firstBagTime + y.firstBagTime, x.traffic + y.traffic, x.timeResponse + y.timeResponse))).name("reduce").uid("db99e12-f90b-400c-b520-8b8ad61a6002")
      .keyBy(_.getKey)
      .process(new AggAndOut(parameterTool)).name("AggAndOut").uid("98d9de4-9bee-4f6e-8abc-0190016ddba7")
      .map(m => {
        (
          //id
          m.host + m.hit + m.httpCode + m.domain + m.source + m.isIpv4 + m.upstreamStatus + m.time,
          //es索引后缀
          TimeUtil.longToString(m.time, "yyyy.MM.dd"),
          JSONUtil.toJsonStr(m)
        )
      })
      .rescale
      .addSink(SinkUtil.esDynamicIndexHaveId(parameterTool)).setParallelism(parameterTool.getInt(sinkParallel, StreamEnv.getStreamExecutionEnvironment.getParallelism)).name("es").uid("9e9993b-c068-4bf1-af7e-331d395bcdf3")
    //      .print("结果数据：")
    StreamEnv.execute(parameterTool)
  }

}
